package com._51doit.sparkTest

import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.tools.scalap.scalax.util.StringUtil

object ETCDemoTest {




  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)
    val sc: SparkContext = new SparkContext(conf)

    val ETLDat: RDD[String] = sc.textFile("D:\\07spark\\ETL\\raw_data.txt")

    var client_type = ""
    var mac_line = ""
    var version = ""
    var dt = ""
    var package_name = ""
    var program_name = ""
    var duration = ""
    var origin = ""
    var total_duration = ""
    var start_time = ""
    var end_time = ""


    val result: RDD[(String, String, String, String, String, String, String, String, String, String, String)] = ETLDat.map(em => {
      if (StringUtils.isEmpty(em)) {
        ("", "", "", "", "", "", "", "", "", "", "")
      } else {
        val logarr: Array[String] = em.split("&")
        try {
          if (logarr.length == 19) {
            dt = logarr(0).split(" ")(0)
            client_type = logarr(3)
            mac_line = logarr(6)
            version = logarr(17)
            package_name = logarr(18).substring(0, logarr(18).indexOf(",", 1)).split("=")(1)
            val arrs: Array[String] = logarr(18).substring(logarr(18).indexOf("{"), logarr(18).indexOf("}") - 1).split(",")
            if (arrs.length > 3) {

              if (arrs.length == 5) {
                origin= arrs(0).split("=")(1)
               program_name= arrs(2).split("=")(1)
                 total_duration= arrs(3).split("=")(1)
               duration = arrs(4).split("=")(1)
              } else {
                var origin = ""
                var program_name = ""
                var duration = ""
                var total_duration = ""
              }

            } else {
              val tem = arrs(0).split("=")(1)
              if (tem.indexOf("-") > -1) {
                start_time = tem
                end_time = arrs(1).split("=")(1)
              } else {
                start_time = ""
                end_time = ""
              }

            }

          } else {
            if (logarr.length == 1) {
              val package_name = logarr(0).substring(0, logarr(0).indexOf(",", 1))
              val arrss: Array[String] = logarr(0).substring(logarr(0).indexOf("{"), logarr(0).indexOf("}") - 1).split(",")
              if (arrss.length > 4) {
                if (arrss.length == 5) {
                  origin = arrss(0).split("=")(1)
                  program_name = arrss(2).split("=")(1)
                  total_duration = arrss(3).split("=")(1)
                  duration = arrss(4).split("=")(1)
                } else {
                  program_name = ""
                  duration = ""
                  origin = ""
                  total_duration = ""
                }
              } else {
                val tem = arrss(0).split("=")(1)
                if (tem.indexOf("-") > -1) {
                  start_time = tem
                  end_time = arrss(1).split("=")(1)
                } else {
                  start_time = ""
                  end_time = ""
                }
              }

            }


          }
        } catch {
          case e: Exception => e.printStackTrace()
        }


        ((client_type, mac_line, package_name, start_time, end_time, program_name, duration, origin, version, total_duration, dt))
      }


    })
    result.take(10)foreach(println)









































































  }



}
